package cn.doit.etl

import cn.doit.bean.AdLog
import cn.doit.common.DolphinAppComm
import cn.doit.utils.FileHelper
import org.apache.spark.rdd.RDD
import org.apache.spark.serializer.KryoSerializer
import org.apache.spark.sql.SQLContext

object Bz2parquetV2 {
  def main(args: Array[String]): Unit = {
    // 检验参数
    if (args.length != 2) {
      println(
        """
          				  |Usage: cn.doit.etl.Bz2parquetV2
          				  |Param:
          				  |	bz2InputPath	bz2日志文件的输入路径
          				  | parquetOutPath	parquet文件的输出路径
        				""".stripMargin)
      sys.exit(-1) // -1 非正常退出
    }

    //接收参数(模式匹配)
    val Array(bz2InputPath,parquetOutPath)=args
    //spark序列化优化参数
    val sparkParams = Map[String, String](
      "spark.serializer" -> classOf[KryoSerializer].getName,
      "spark.sql.parquet.compression.code" -> "snappy" //设置parquet文件的压缩格式
    )


    val sc = DolphinAppComm.creatSparkContext("将bz2日志文件转换成parquet文件",sparkParams)

  //读取离线的bz2日志文件
    val file = sc.textFile(bz2InputPath)
  //过滤非法数据
    val filterRDD: RDD[Array[String]] = file.map(_.split(",",-1)).filter(_.size>=85)
  //parquet <- DataFrame(几种创建方式) <- SQLContext <- RDD
    val sqlc = new SQLContext(sc)
  //RDD[Array[String]] -> RDD[AdLog]
    val adLogRDD: RDD[AdLog] = filterRDD.map(AdLog(_))
    val dataFrame = sqlc.createDataFrame(adLogRDD)

    //删除目标目录
   FileHelper.deleteDir(parquetOutPath,sc)
    //parrtitionBy 对数据按照省份进行区分

    dataFrame.write.parquet(parquetOutPath)
    sc.stop()
  }

}
